package com.atguigu.flink.state.opearatestate;

import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.state.ListState;
import org.apache.flink.api.common.state.ListStateDescriptor;
import org.apache.flink.api.common.state.OperatorStateStore;
import org.apache.flink.runtime.state.FunctionInitializationContext;
import org.apache.flink.runtime.state.FunctionSnapshotContext;
import org.apache.flink.streaming.api.checkpoint.CheckpointedFunction;
import org.apache.flink.streaming.api.environment.CheckpointConfig;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.sink.SinkFunction;

/**
 * Created by 黄凯 on 2023/6/20 0020 18:18
 *
 * @author 黄凯
 * 永远相信美好的事情总会发生.
 * <p>
 * ManagedState: 管理状态。使用Flink提供的状态功能定义的状态。Flink自动维护状态。
 * *          KeyedState:  KeyedStream 上所有的Task使用的状态都是KeyedState。keyBy
 * *                              每个Key都有自己的状态。
 * *
 * *          OpearateState: 普通的流 DataStream。 不keyBy。
 * *                              每个Task上共用一个状态。
 * *                              让Task实现 CheckpointedFunction
 */
public class Flink02_ListState {

    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        env.setParallelism(2);

        //备份的配置信息
        CheckpointConfig checkpointConfig = env.getCheckpointConfig();

        //开启了checkpoint(备份)，每间隔2s自动备份，程序就会无限次重启
        env.enableCheckpointing(2000);

        env.socketTextStream("127.0.0.1", 8888)
                .map(new MyMapFunction())
                .addSink(
                        new SinkFunction<String>() {
                            @Override
                            public void invoke(String value, Context context) throws Exception {

                                if (value.contains("x")) {

                                    //人为手动抛出一个异常，为了模拟故障的情况
                                    throw new RuntimeException("出异常了！！！");

                                }

                                System.out.println(value);

                            }
                        }
                );

        env.execute();

    }

    /**
     * 功能：实现字符串的累积打印
     * <p>
     * flink提供的状态就是拥有自动备份，自动恢复功能的集合
     */
    public static class MyMapFunction implements MapFunction<String, String>, CheckpointedFunction {


        //当集合用，本质是managerState中的OpearateState
        ListState<String> listState1;
        ListState<String> listState2;


        @Override
        public String map(String value) throws Exception {

            //存
            listState1.add(value);

            //取
            return listState1.get().toString();


        }

        /**
         * 周期性（默认200ms，可以调整）将状态以快照的形式进行备份
         *
         * @param context the context for drawing a snapshot of the operator
         * @throws Exception
         */
        @Override
        public void snapshotState(FunctionSnapshotContext context) throws Exception {

            System.out.println("MyMapFunction.snapshotState");

            //备份是自动进行的，无需进行任何手动操作

        }

        /**
         * 初始化，在第一次启动或任务失败重启后执行，从之前的备份中找到状态，重新赋值
         * <p>
         * FunctionInitializationContext context：程序的运行环境，或上下文，可以从中获取很多额外的信息
         */
        @Override
        public void initializeState(FunctionInitializationContext context) throws Exception {

            System.out.println("MyMapFunction.initializeState");

            //找到备份状态的存储设备
            OperatorStateStore operatorStateStore = context.getOperatorStateStore();

            //从备份中找到之前备份的变量
            ListStateDescriptor<String> stateDescriptor1 = new ListStateDescriptor<String>("list1", String.class);
            ListStateDescriptor<String> stateDescriptor2 = new ListStateDescriptor<String>("list2", String.class);

            listState1 = operatorStateStore.getListState(stateDescriptor1);
            listState2 = operatorStateStore.getListState(stateDescriptor2);

        }
    }

}
